package com.cn.broker;

import com.cn.common.ServerThread;

import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 挂起请求
 */
public class PullMessageRequestHoldService extends ServerThread {


    private ConcurrentMap<String /* topic-queueid */, ManyPullRequest> pullRequestTables = new ConcurrentHashMap<>();

    private BrokerController brokerController;

    public PullMessageRequestHoldService(BrokerController brokerController) {
        this.brokerController = brokerController;
    }


    @Override
    public String getServiceName() {
        return PullMessageRequestHoldService.class.getSimpleName();
    }

    public String buildKey(PullRequest pullRequest) {
        return pullRequest.getTopic() + "-" + pullRequest.getQueueId();
    }

    public void suspendRequest(PullRequest pullRequest) {
        String key = buildKey(pullRequest);
        ManyPullRequest mq = this.pullRequestTables.get(key);
        if (mq == null) {
            mq = new ManyPullRequest();
            this.pullRequestTables.put(key, mq);
        }
        if (mq != null) {
            mq.add(pullRequest);
        }
    }

    public void checkHoldRequest() {
        //遍历hold请求
        for (String key : this.pullRequestTables.keySet()) {
            String[] ids = key.split("-");
            if (ids.length == 2) {
                //查询是否有新的消息到达
                long offset = this.brokerController.getDefaultMessageStore().findConsumeQueue(ids[0], Integer.valueOf(ids[1])).getConsumeQueueOffset();
                notifyMessageArriving(key, ids[0], Integer.valueOf(ids[1]), offset);
            }
        }
    }

    public void notifyMessageArriving(String key, String topic, int queueId, long offset) {
        ManyPullRequest mq = this.pullRequestTables.get(key);
        if (mq != null) {
            ArrayList<PullRequest> requestList = mq.copyPullRequest();
            if (requestList == null) {
                return;
            }
            ArrayList<PullRequest> reReqList = new ArrayList<>();
            for (PullRequest request : requestList) {
                long oldOffset = request.getOffset();
                if (offset > oldOffset) {
                    //有新的消息来了
                    System.out.println("有新的消息来了：" + offset);
                    this.brokerController.getPullProcessor().wakeupHoldRequest(request.getChannel(), request.getMessage(), request);
                    continue;
                }

                //挂起时间到期了
                if (System.currentTimeMillis() >= (request.getSuspendTimeStamp() + this.brokerController.getPullSuspendTimeStamp())) {
                    System.out.println(String.format("主题：%s,队列：%d，id:%d, 等待超时。", request.getTopic(), request.getQueueId(), request.getMessage().getOpaque()));
                    this.brokerController.getPullProcessor().wakeupHoldRequest(request.getChannel(), request.getMessage(), request);
                    continue;
                }
                reReqList.add(request);
            }

            if (!reReqList.isEmpty()) {
                mq.addAll(reReqList);
            }
        }
    }

    @Override
    public void run() {
        while (!isStopped()) {
            //等待5秒
            this.waitForRunning(5000);
            long start = System.currentTimeMillis();
            checkHoldRequest();
//            long take = System.currentTimeMillis() - start;
//            System.out.println("花费时间：" + take + "秒");
        }
    }
}
